package com.cssl.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cssl.bean.OrderWide;
import com.cssl.bean.PaymentInfo;
import com.cssl.bean.PaymentWide;
import com.cssl.utils.CommonUtils;
import com.cssl.utils.DateTimeUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/16 22:57
 * @Desc: join paymentInfo & orderWide
 */
public class PaymentWideApp {
    public static void main(String[] args) throws Exception {
        //1. 获取执行环境
        Properties properties = CommonUtils.getProperties();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 读取 kafka 主题的数据创建流 并转换为JavaBean对象 提取时间戳生成 watermark
        //2.1 orderWide
        String orderWideSourceTopic = properties.getProperty("kafka.topic.db.dwm.orderWide");
        SingleOutputStreamOperator<OrderWide> orderWideDS = env.addSource(CommonUtils.getKafkaConsumer(orderWideSourceTopic))
                .map(line -> JSON.parseObject(line, OrderWide.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderWide>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<OrderWide>() {
                    @Override
                    public long extractTimestamp(OrderWide element, long recordTimestamp) {
                        return DateTimeUtil.toTs(element.getCreate_time());
                    }
                }));

        //2.2paymentInfo
        String paymentInfoSourceTopic = properties.getProperty("kafka.topic.db.dwd.paymentInfo");
        SingleOutputStreamOperator<PaymentInfo> paymentInfoDS = env.addSource(CommonUtils.getKafkaConsumer(paymentInfoSourceTopic))
                .map(line -> JSON.parseObject(line, PaymentInfo.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<PaymentInfo>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<PaymentInfo>() {
                    @Override
                    public long extractTimestamp(PaymentInfo element, long recordTimestamp) {
                        return DateTimeUtil.toTs(element.getCreate_time());
                    }
                }));

        //3. 双流 join
        SingleOutputStreamOperator<PaymentWide> paymentWideDS = paymentInfoDS.keyBy(PaymentInfo::getOrder_id)
                .intervalJoin(orderWideDS.keyBy(OrderWide::getOrder_id))
                .between(Time.minutes(-15), Time.seconds(5))
                .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                    @Override
                    public void processElement(PaymentInfo left, OrderWide right, Context ctx, Collector<PaymentWide> out) throws Exception {
                        out.collect(new PaymentWide(left, right));
                    }
                });

        //4. 将数据写入 kafka
        String paymentWideSinkTopic = properties.getProperty("kafka.topic.db.dwm.paymentWide");
        paymentWideDS.map(JSONObject::toJSONString).addSink(CommonUtils.getKafkaProducer(paymentWideSinkTopic));

        //5. 启动任务
        env.execute("PaymentWideApp");
    }
}
